/*
 * Copyright 2016 NAVER Corp.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.ultrapower.mapper;

import org.apache.hadoop.hbase.client.Result;

import com.ultrapower.bean.AgentInfoBean;
import com.ultrapower.bean.JvmInfoBean;
import com.ultrapower.bean.ServerMetaDataBean;
import com.ultrapower.buffer.Buffer;
import com.ultrapower.buffer.FixedBuffer;
import com.ultrapower.hbase.HbaseUtil;
import com.ultrapower.util.BytesUtils;
import com.ultrapower.util.HBaseTables;

/**
 * @author HyunGil Jeong
 */
public class AgentInfoMapper implements RowMapper<AgentInfoBean> {

    @Override
    public AgentInfoBean mapRow(Result result, int rowNum) throws Exception {

        byte[] rowKey = result.getRow();
        String agentId = BytesUtils.safeTrim(BytesUtils.toString(rowKey, 0, HBaseTables.AGENT_NAME_MAX_LEN));
        long reverseStartTime = BytesUtils.bytesToLong(rowKey, HBaseTables.AGENT_NAME_MAX_LEN);
        long startTime = HbaseUtil.recoveryTimeMillis(reverseStartTime);

        byte[] serializedAgentInfo = result.getValue(HBaseTables.AGENTINFO_CF_INFO, HBaseTables.AGENTINFO_CF_INFO_IDENTIFIER);
        byte[] serializedServerMetaData = result.getValue(HBaseTables.AGENTINFO_CF_INFO, HBaseTables.AGENTINFO_CF_INFO_SERVER_META_DATA);
        byte[] serializedJvmInfo = result.getValue(HBaseTables.AGENTINFO_CF_INFO, HBaseTables.AGENTINFO_CF_INFO_JVM);

        AgentInfoBean agentInfo = createBuilderFromValue(serializedAgentInfo);
        agentInfo.setAgentId(agentId);
        agentInfo.setStartTimestamp(startTime);

        if (serializedServerMetaData != null) {
            agentInfo.setServerMetaData(new ServerMetaDataBean(serializedServerMetaData));
        }
        if (serializedJvmInfo != null) {
        	agentInfo.setJvmInfo(new JvmInfoBean(serializedJvmInfo));
        }
        return agentInfo;
    }

    private AgentInfoBean createBuilderFromValue(byte[] serializedAgentInfo) {
        final Buffer buffer = new FixedBuffer(serializedAgentInfo);
        AgentInfoBean agentInfo=new AgentInfoBean();
        agentInfo.setHostName(buffer.readPrefixedString());
        agentInfo.setIp(buffer.readPrefixedString());
        agentInfo.setPorts(buffer.readPrefixedString());
        agentInfo.setApplicationName(buffer.readPrefixedString());
        agentInfo.setServiceTypeCode(buffer.readShort());
        agentInfo.setPid(buffer.readInt());
        agentInfo.setAgentVersion(buffer.readPrefixedString());
        //agentInfo.setStartTimestamp(buffer.readLong());
//        agentInfo.setEndTimeStamp(buffer.readLong());
//        agentInfo.setEndStatus(buffer.readInt());
        // FIXME - 2015.09 v1.5.0 added vmVersion (check for compatibility)
        if (buffer.hasRemaining()) {
            agentInfo.setVmVersion(buffer.readPrefixedString());
        }
        return agentInfo;
    }
}
